package com.moli.iov.handle.real.mq.kafka;


import com.moli.iov.cache.RedisUtil;
import com.moli.iov.consumer.KafkaConsumerServer;
import com.moli.iov.down.remote.IVehicleCommandRemoteService;
import com.moli.iov.handle.real.service.IVehicleService;
import com.moli.iov.protocol.codec.DataDecoder;
import com.moli.iov.protocol.codec.ReqStreamData;
import com.moli.iov.protocol.enums.ResponseTag;
import com.moli.iov.protocol.model.VehicleRegisterReq;
import com.moli.iov.rest.RestResponse;
import com.moli.iov.util.ByteArrayUtil;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;


/**
 * kafka 注册报文消费者
 */
@Data
@Slf4j
public class RegisterConsumer extends KafkaConsumerServer {

    //redis封装工具类
    @Resource
    RedisUtil<String, String> redisUtil;

    @Resource
    IVehicleService vehicleService;


    @Resource
    IVehicleCommandRemoteService vehicleCommandRemoteService;

    //kafka数据处理类
    @Override
    public void manualHandle(ConsumerRecord<String, Object> record,
                             Acknowledgment acknowledgment) {
        long start = System.currentTimeMillis();
        try {
            //将kafka数据解析为报文通用对象
            ReqStreamData data = (ReqStreamData) record.value();
            if (data != null) {
                VehicleRegisterReq registerReq = new VehicleRegisterReq();

                //解析ReqStreamData中的注册报文
                DataDecoder<VehicleRegisterReq> decoder = new DataDecoder<VehicleRegisterReq>();
                decoder.decode(registerReq, data.getDataBytes());
                log.info("[车辆注册报文]{}", registerReq.toString());

                //调用基础服务向车联网注册车联信息
                register(registerReq);
//			redisUtil.hset(RealContant.KEY_MSG_REALTIME, record.key(),buildJson(data.getCollectDate(), realTimeReq));
            } else {
                log.info("[注册报文][ReqStreamData 为空]");
            }
        } catch (Exception e) {
            log.error("[车辆注册][decode error][vin={}][data={}][err={}]", record.key(), record.value(), e.getMessage());
        }
        long time = System.currentTimeMillis() - start;
        log.info("[车辆:{}，注册:{}][时长:{}ms]",record.key(),record.value(),time);
        acknowledgment.acknowledge();
    }

    /**
     * 调用基础服务向车联网注册车联信息
     *
     * @param registerReq 注册报文
     */
    private void register(VehicleRegisterReq registerReq) {

        if (registerReq == null || StringUtils.isEmpty(registerReq.getDeviceNum())
                || StringUtils.isEmpty(registerReq.getIccid()) || StringUtils.isEmpty(registerReq.getVin())) {
            log.info("[注册报文][registerReq 数据异常][{}]", registerReq);
            return;
        }

        byte[] serialNumber = ByteArrayUtil.toBytes(registerReq.getSerialNum());
        Byte resByte =null;

        try {
            if (vehicleService.registerVehicle(registerReq)) {
                //执行成功
                log.info("[注册报文][入库成功][{}]", registerReq.toString());

                resByte=ResponseTag.SUCCESS.getV();
            } else {
                //执行失败
                log.info("[注册报文][入库失败][{}]", registerReq.toString());
//            注册失败应答接口
                resByte=ResponseTag.ERROR.getV();
            }
        }catch (Exception e){
            log.error("[注册报文][入库失败][{}][mess:{}]", registerReq.toString(),e.getMessage());
            e.printStackTrace();
        }

        try{
            RestResponse res = vehicleCommandRemoteService.sendResponse(registerReq.getVin(), ResponseTag.REGISTER_RES.getV(), resByte, serialNumber);
            if(res==null || res!=null && !res.isFlag()){
                log.error("[注册报文][发送应答失败][{}]", registerReq.toString());
            }else{
                log.info("[注册报文][发送应答成功][{}]", registerReq.toString());
            }
        }catch(Exception e){
            log.error("[注册报文][发送应答失败][param:{}][mess:{}]", registerReq.toString(),e.getMessage());
            e.printStackTrace();
        }

    }


}
